package com.fanxl.flink.datastream.watermark;


import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;

import javax.annotation.Nullable;

/**
 * todo 水平线的使用，一直需要认真理解原理
 *
 * @author Administrator
 * @Title:
 * @Description: 数据源处理类
 * @date 2021/11/25
 */
public class TuplePeriodicWatermark implements AssignerWithPeriodicWatermarks<Tuple2<String, Long>> {

    private boolean debug;

    public TuplePeriodicWatermark(boolean debug) {
        this.debug = debug;
    }

    private long currentTimestamp = Long.MIN_VALUE;

    @Override
    public long extractTimestamp(Tuple2<String, Long > word, long previousElementTimestamp) {
        long timestamp =  word.f1;
        currentTimestamp = Math.max(timestamp, currentTimestamp);
        return  timestamp;
    }

    @Nullable
    @Override
    public Watermark getCurrentWatermark() {
        long maxTimeLag = 1000;
        Watermark watermark = new Watermark(currentTimestamp == Long.MIN_VALUE ? Long.MIN_VALUE : currentTimestamp - maxTimeLag);
        return watermark;
    }


}
